package com.heima.schedule.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.nacos.client.naming.utils.CollectionUtils;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.heima.common.cache.CacheService;
import com.heima.common.constants.ScheduleConstants;
import com.heima.model.schedule.dto.Task;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Date;
import java.util.List;
import java.util.Set;


@Service
@Slf4j
public class TaskServiceImpl implements TaskService {

    @Autowired
    private TaskinfoService taskinfoService;

    @Autowired
    private TaskinfoLogsService taskinfoLogsService;

    @Autowired
    private CacheService cacheService;
    
    @Override
    @Transactional(rollbackFor = RuntimeException.class)
    public long addTask(Task task) {
        // 添加任务到task
        long executeTime = task.getExecuteTime();

        Taskinfo taskInfo = new Taskinfo();
        BeanUtils.copyProperties(task, taskInfo);
        taskInfo.setExecuteTime(new Date(executeTime));

        boolean taskResult = taskinfoService.save(taskInfo);
        if (!taskResult) {
            log.warn("任务添加失败");
            throw new RuntimeException("任务添加失败");
        }

        // 添加任务到log
        TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
        BeanUtils.copyProperties(task, log);
        taskinfoLogs.setExecuteTime(new Date(executeTime));
        taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED);
//        taskinfoLogs.setVersion(1);
        boolean logResult = taskinfoLogsService.save(taskinfoLogs);
        if (!logResult) {
            log.warn("任务日志添加失败");
            throw new RuntimeException("任务日志添加失败");
        }

        addTaskToCache(task);

        // 返回任务id
        return taskInfo.getTaskId();
    }


    public void addTaskToCache(Task task) {
        long executeTime = task.getExecuteTime();
        long currentTimeMillis = System.currentTimeMillis();
        long futureTime = currentTimeMillis + (5 * 60 * 1000);
        // 如果小于当前就写入list
        String listKey = ScheduleConstants.TOPIC;
        // 如果小于五分钟后，就写入zset
        String zSetKey = ScheduleConstants.FUTURE;
        if (executeTime <= currentTimeMillis) {
            cacheService.lLeftPush(listKey, JSON.toJSONString(task));
        } else if (executeTime <= futureTime) {
            cacheService.zAdd(zSetKey, JSON.toJSONString(task), executeTime);
        }
    }
    @Override
    public Boolean cancelTask(Long taskId) {
  //入参判空
        if (taskId==null){
            log.warn("参数不能为空");
            return false;
        }
     //删除taskinfolog
        boolean removeResult = taskinfoService.removeById(taskId);
        if (!removeResult){
            log.warn("删除失败");
            return false;
        }
//更新taskinfolog
        LambdaQueryWrapper<TaskinfoLogs> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(TaskinfoLogs::getTaskId,taskId);
        TaskinfoLogs logs = taskinfoLogsService.getOne(wrapper);
        logs.setStatus(2);

        boolean updateResult = taskinfoLogsService.updateById(logs);
        if (!updateResult){
            log.warn("更新失败");
            throw new RuntimeException("日志更新失败");
        }
        //组装task数据
        Task task = new Task();
        BeanUtils.copyProperties(logs,task);
        task.setExecuteTime(logs.getExecuteTime().getTime());
        cacheService.lRemove("TOPIC",0,JSON.toJSONString(task));
        cacheService.zRemove("FUTURE",JSON.toJSONString(task));
        return true;
    }

    @Override
    @Transactional(rollbackFor = RuntimeException.class)
    public Boolean pollTask() {
        //从topic中弹出数据
        String topic = cacheService.lRightPop("TOPIC");
        if (StringUtils.isEmpty(topic)){
            log.warn("无任务执行");
            return false;
        }
        //拿到数据组装成task对象
        Task task = JSON.parseObject(topic, Task.class);
        if (task==null){
            log.warn("无执行任务");
            return false;
        }
        boolean removeById = taskinfoService.removeById(task.getTaskId());
        if (!removeById){
            log.warn("删除失败");
            return false;
        }
        long executeTime = task.getExecuteTime();
        TaskinfoLogs logs = new TaskinfoLogs();
        BeanUtils.copyProperties(task,logs);
        logs.setExecuteTime(new Date(executeTime));
        logs.setStatus(1);
        //更新
        boolean updateById = taskinfoLogsService.updateById(logs);
        if (!updateById){
            log.warn("更新失败");
            throw new RuntimeException("更新失败");
        }
return true;
    }
    /**
     * 定时刷新任务
     */
    //***是指秒分小时天月
//    @Scheduled(cron = "0 */1 * * * ?")
//    public void refresh() {
//        System.out.println(System.currentTimeMillis() / 1000 + "执行了定时任务");
//
//        // 获取该组key下当前需要消费的任务数据
//        // 获取0~当前时间的所有任务（代表小于当前时间的任务）
//        // 任务a，执行时间100000,0 ~ 100020
//        // 任务b，执行时间100300
//        Set<String> tasks = cacheService.zRangeByScore(ScheduleConstants.FUTURE, 0, System.currentTimeMillis());
//        if (!tasks.isEmpty()) {
//            //将这些任务数据添加到消费者队列中
//            cacheService.refreshWithPipeline(ScheduleConstants.FUTURE, ScheduleConstants.TOPIC, tasks);
//            System.out.println("成功的将" + ScheduleConstants.FUTURE + "下的当前需要执行的任务数据刷新到" + ScheduleConstants.TOPIC + "下");
//        }
//    }


    @Scheduled(fixedRate = 5 * 60 * 1000)
    public void initData() {
        // 清除缓存
        clear();
//查询taskinfo里所有数据
        List<Taskinfo> list = taskinfoService.list();
        //先对集合判断是否为空然后再循环
        if (CollectionUtils.isEmpty(list)) {
            log.warn("任务为空");
            return;
        }

        for (Taskinfo taskinfo : list) {
            if (taskinfo == null) {
                continue;
            }

            //获取taskinfo时间戳
            //获取当前时间
            //获取未来时间（5分钟时间)
            long executeTime = taskinfo.getExecuteTime().getTime();
            long currentTime = System.currentTimeMillis();
            long futureTime = currentTime + (5 * 60 * 1000);

            Task task = new Task();
            BeanUtils.copyProperties(taskinfo, task);
            task.setExecuteTime(executeTime);

            //判断执行时间是否小于当前时间
            if (executeTime <= currentTime) {
                cacheService.lLeftPush("TOPIC", JSON.toJSONString(task));
            } else if (executeTime <= futureTime) {//判断执行时间是否小于未来 时间
                cacheService.zAdd("FUTURE", JSON.toJSONString(task), executeTime);
            }
        }
    }

    private void clear() {
        cacheService.delete("TOPIC");
        cacheService.delete("FUTURE");
    }

    @Scheduled(fixedRate = 5*60*1000)
    public void refresh(){
        log.info("我被执行啦");
        //每隔五分钟上一次锁
        String refresh_lock = cacheService.tryLock("refresh_lock", 5000);
        if (refresh_lock==null){
            return;
        }
        Set<String> future = cacheService.zRangeByScore("FUTURE", 0, System.currentTimeMillis());
        if (CollectionUtils.isEmpty(future)){
            log.warn("无可执行任务");
        }
cacheService.refreshWithPipeline("FUTURE","TOPIC",future);
    }
}